3 use MediaWiki\MediaWikiServices
;
10 class JobQueueTest
extends MediaWikiTestCase
{
12 protected $queueRand, $queueRandTTL, $queueFifo, $queueFifoTTL;
14 function __construct( $name = null, array $data = [], $dataName = '' ) {
15 parent
::__construct( $name, $data, $dataName );
17 $this->tablesUsed
[] = 'job';
20 protected function setUp() {
21 global $wgJobTypeConf;
24 if ( $this->getCliArg( 'use-jobqueue' ) ) {
25 $name = $this->getCliArg( 'use-jobqueue' );
26 if ( !isset( $wgJobTypeConf[$name] ) ) {
27 throw new MWException( "No \$wgJobTypeConf entry for '$name'." );
29 $baseConfig = $wgJobTypeConf[$name];
31 $baseConfig = [ 'class' => JobQueueDBSingle
::class ];
33 $baseConfig['type'] = 'null';
34 $baseConfig['wiki'] = wfWikiID();
36 'queueRand' => [ 'order' => 'random', 'claimTTL' => 0 ],
37 'queueRandTTL' => [ 'order' => 'random', 'claimTTL' => 10 ],
38 'queueTimestamp' => [ 'order' => 'timestamp', 'claimTTL' => 0 ],
39 'queueTimestampTTL' => [ 'order' => 'timestamp', 'claimTTL' => 10 ],
40 'queueFifo' => [ 'order' => 'fifo', 'claimTTL' => 0 ],
41 'queueFifoTTL' => [ 'order' => 'fifo', 'claimTTL' => 10 ],
43 foreach ( $variants as $q => $settings ) {
45 $this->$q = JobQueue
::factory( $settings +
$baseConfig );
46 } catch ( MWException
$e ) {
48 // @todo What if it was another error?
53 protected function tearDown() {
57 'queueRand', 'queueRandTTL', 'queueTimestamp', 'queueTimestampTTL',
58 'queueFifo', 'queueFifoTTL'
69 * @dataProvider provider_queueLists
70 * @covers JobQueue::getWiki
72 public function testGetWiki( $queue, $recycles, $desc ) {
73 $queue = $this->$queue;
75 $this->markTestSkipped( $desc );
77 $this->assertEquals( wfWikiID(), $queue->getWiki(), "Proper wiki ID ($desc)" );
81 * @dataProvider provider_queueLists
82 * @covers JobQueue::getType
84 public function testGetType( $queue, $recycles, $desc ) {
85 $queue = $this->$queue;
87 $this->markTestSkipped( $desc );
89 $this->assertEquals( 'null', $queue->getType(), "Proper job type ($desc)" );
93 * @dataProvider provider_queueLists
96 public function testBasicOperations( $queue, $recycles, $desc ) {
97 $queue = $this->$queue;
99 $this->markTestSkipped( $desc );
102 $this->assertTrue( $queue->isEmpty(), "Queue is empty ($desc)" );
104 $queue->flushCaches();
105 $this->assertEquals( 0, $queue->getSize(), "Queue is empty ($desc)" );
106 $this->assertEquals( 0, $queue->getAcquiredCount(), "Queue is empty ($desc)" );
108 $this->assertNull( $queue->push( $this->newJob() ), "Push worked ($desc)" );
109 $this->assertNull( $queue->batchPush( [ $this->newJob() ] ), "Push worked ($desc)" );
111 $this->assertFalse( $queue->isEmpty(), "Queue is not empty ($desc)" );
113 $queue->flushCaches();
114 $this->assertEquals( 2, $queue->getSize(), "Queue size is correct ($desc)" );
115 $this->assertEquals( 0, $queue->getAcquiredCount(), "No jobs active ($desc)" );
116 $jobs = iterator_to_array( $queue->getAllQueuedJobs() );
117 $this->assertEquals( 2, count( $jobs ), "Queue iterator size is correct ($desc)" );
119 $job1 = $queue->pop();
120 $this->assertFalse( $queue->isEmpty(), "Queue is not empty ($desc)" );
122 $queue->flushCaches();
123 $this->assertEquals( 1, $queue->getSize(), "Queue size is correct ($desc)" );
125 $queue->flushCaches();
127 $this->assertEquals( 1, $queue->getAcquiredCount(), "Active job count ($desc)" );
130 $job2 = $queue->pop();
131 $this->assertTrue( $queue->isEmpty(), "Queue is empty ($desc)" );
132 $this->assertEquals( 0, $queue->getSize(), "Queue is empty ($desc)" );
134 $queue->flushCaches();
136 $this->assertEquals( 2, $queue->getAcquiredCount(), "Active job count ($desc)" );
139 $queue->ack( $job1 );
141 $queue->flushCaches();
143 $this->assertEquals( 1, $queue->getAcquiredCount(), "Active job count ($desc)" );
146 $queue->ack( $job2 );
148 $queue->flushCaches();
149 $this->assertEquals( 0, $queue->getAcquiredCount(), "Active job count ($desc)" );
151 $this->assertNull( $queue->batchPush( [ $this->newJob(), $this->newJob() ] ),
152 "Push worked ($desc)" );
153 $this->assertFalse( $queue->isEmpty(), "Queue is not empty ($desc)" );
156 $queue->flushCaches();
157 $this->assertTrue( $queue->isEmpty(), "Queue is empty ($desc)" );
158 $this->assertEquals( 0, $queue->getSize(), "Queue is empty ($desc)" );
162 * @dataProvider provider_queueLists
165 public function testBasicDeduplication( $queue, $recycles, $desc ) {
166 $queue = $this->$queue;
168 $this->markTestSkipped( $desc );
171 $this->assertTrue( $queue->isEmpty(), "Queue is empty ($desc)" );
173 $queue->flushCaches();
174 $this->assertEquals( 0, $queue->getSize(), "Queue is empty ($desc)" );
175 $this->assertEquals( 0, $queue->getAcquiredCount(), "Queue is empty ($desc)" );
179 [ $this->newDedupedJob(), $this->newDedupedJob(), $this->newDedupedJob() ]
181 "Push worked ($desc)" );
183 $this->assertFalse( $queue->isEmpty(), "Queue is not empty ($desc)" );
185 $queue->flushCaches();
186 $this->assertEquals( 1, $queue->getSize(), "Queue size is correct ($desc)" );
187 $this->assertEquals( 0, $queue->getAcquiredCount(), "No jobs active ($desc)" );
191 [ $this->newDedupedJob(), $this->newDedupedJob(), $this->newDedupedJob() ]
193 "Push worked ($desc)"
196 $this->assertFalse( $queue->isEmpty(), "Queue is not empty ($desc)" );
198 $queue->flushCaches();
199 $this->assertEquals( 1, $queue->getSize(), "Queue size is correct ($desc)" );
200 $this->assertEquals( 0, $queue->getAcquiredCount(), "No jobs active ($desc)" );
202 $job1 = $queue->pop();
203 $this->assertTrue( $queue->isEmpty(), "Queue is empty ($desc)" );
205 $queue->flushCaches();
206 $this->assertEquals( 0, $queue->getSize(), "Queue is empty ($desc)" );
208 $this->assertEquals( 1, $queue->getAcquiredCount(), "Active job count ($desc)" );
211 $queue->ack( $job1 );
213 $queue->flushCaches();
214 $this->assertEquals( 0, $queue->getAcquiredCount(), "Active job count ($desc)" );
218 * @dataProvider provider_queueLists
221 public function testDeduplicationWhileClaimed( $queue, $recycles, $desc ) {
222 $queue = $this->$queue;
224 $this->markTestSkipped( $desc );
227 $job = $this->newDedupedJob();
228 $queue->push( $job );
230 // De-duplication does not apply to already-claimed jobs
232 $queue->push( $job );
236 // Make sure ack() of the twin did not delete the sibling data
237 $this->assertType( NullJob
::class, $j );
241 * @dataProvider provider_queueLists
244 public function testRootDeduplication( $queue, $recycles, $desc ) {
245 $queue = $this->$queue;
247 $this->markTestSkipped( $desc );
250 $this->assertTrue( $queue->isEmpty(), "Queue is empty ($desc)" );
252 $queue->flushCaches();
253 $this->assertEquals( 0, $queue->getSize(), "Queue is empty ($desc)" );
254 $this->assertEquals( 0, $queue->getAcquiredCount(), "Queue is empty ($desc)" );
256 $id = wfRandomString( 32 );
257 $root1 = Job
::newRootJobParams( "nulljobspam:$id" ); // task ID/timestamp
258 for ( $i = 0; $i < 5; ++
$i ) {
259 $this->assertNull( $queue->push( $this->newJob( 0, $root1 ) ), "Push worked ($desc)" );
261 $queue->deduplicateRootJob( $this->newJob( 0, $root1 ) );
264 # Add a second to UNIX epoch and format back to TS_MW
265 $root2_ts = strtotime( $root2['rootJobTimestamp'] );
267 $root2['rootJobTimestamp'] = wfTimestamp( TS_MW
, $root2_ts );
269 $this->assertNotEquals( $root1['rootJobTimestamp'], $root2['rootJobTimestamp'],
270 "Root job signatures have different timestamps." );
271 for ( $i = 0; $i < 5; ++
$i ) {
272 $this->assertNull( $queue->push( $this->newJob( 0, $root2 ) ), "Push worked ($desc)" );
274 $queue->deduplicateRootJob( $this->newJob( 0, $root2 ) );
276 $this->assertFalse( $queue->isEmpty(), "Queue is not empty ($desc)" );
278 $queue->flushCaches();
279 $this->assertEquals( 10, $queue->getSize(), "Queue size is correct ($desc)" );
280 $this->assertEquals( 0, $queue->getAcquiredCount(), "No jobs active ($desc)" );
285 $job = $queue->pop();
290 if ( $job instanceof DuplicateJob
) {
295 $this->assertEquals( 10, count( $jobs ), "Correct number of jobs popped ($desc)" );
296 $this->assertEquals( 5, $dupcount, "Correct number of duplicate jobs popped ($desc)" );
300 * @dataProvider provider_fifoQueueLists
303 public function testJobOrder( $queue, $recycles, $desc ) {
304 $queue = $this->$queue;
306 $this->markTestSkipped( $desc );
309 $this->assertTrue( $queue->isEmpty(), "Queue is empty ($desc)" );
311 $queue->flushCaches();
312 $this->assertEquals( 0, $queue->getSize(), "Queue is empty ($desc)" );
313 $this->assertEquals( 0, $queue->getAcquiredCount(), "Queue is empty ($desc)" );
315 for ( $i = 0; $i < 10; ++
$i ) {
316 $this->assertNull( $queue->push( $this->newJob( $i ) ), "Push worked ($desc)" );
319 for ( $i = 0; $i < 10; ++
$i ) {
320 $job = $queue->pop();
321 $this->assertTrue( $job instanceof Job
, "Jobs popped from queue ($desc)" );
322 $params = $job->getParams();
323 $this->assertEquals( $i, $params['i'], "Job popped from queue is FIFO ($desc)" );
327 $this->assertFalse( $queue->pop(), "Queue is not empty ($desc)" );
329 $queue->flushCaches();
330 $this->assertEquals( 0, $queue->getSize(), "Queue is empty ($desc)" );
331 $this->assertEquals( 0, $queue->getAcquiredCount(), "No jobs active ($desc)" );
337 public function testQueueAggregateTable() {
338 $queue = $this->queueFifo
;
339 if ( !$queue ||
!method_exists( $queue, 'getServerQueuesWithJobs' ) ) {
340 $this->markTestSkipped();
343 $this->assertNotContains(
344 [ $queue->getType(), $queue->getWiki() ],
345 $queue->getServerQueuesWithJobs(),
346 "Null queue not in listing"
349 $queue->push( $this->newJob( 0 ) );
351 $this->assertContains(
352 [ $queue->getType(), $queue->getWiki() ],
353 $queue->getServerQueuesWithJobs(),
354 "Null queue in listing"
358 public static function provider_queueLists() {
360 [ 'queueRand', false, 'Random queue without ack()' ],
361 [ 'queueRandTTL', true, 'Random queue with ack()' ],
362 [ 'queueTimestamp', false, 'Time ordered queue without ack()' ],
363 [ 'queueTimestampTTL', true, 'Time ordered queue with ack()' ],
364 [ 'queueFifo', false, 'FIFO ordered queue without ack()' ],
365 [ 'queueFifoTTL', true, 'FIFO ordered queue with ack()' ]
369 public static function provider_fifoQueueLists() {
371 [ 'queueFifo', false, 'Ordered queue without ack()' ],
372 [ 'queueFifoTTL', true, 'Ordered queue with ack()' ]
376 function newJob( $i = 0, $rootJob = [] ) {
377 return new NullJob( Title
::newMainPage(),
378 [ 'lives' => 0, 'usleep' => 0, 'removeDuplicates' => 0, 'i' => $i ] +
$rootJob );
381 function newDedupedJob( $i = 0, $rootJob = [] ) {
382 return new NullJob( Title
::newMainPage(),
383 [ 'lives' => 0, 'usleep' => 0, 'removeDuplicates' => 1, 'i' => $i ] +
$rootJob );
387 class JobQueueDBSingle
extends JobQueueDB
{
388 protected function getDB( $index ) {
389 $lb = MediaWikiServices
::getInstance()->getDBLoadBalancer();
390 // Override to not use CONN_TRX_AUTOCOMMIT so that we see the same temporary `job` table
391 return $lb->getConnection( $index, [], $this->wiki
);